package conf

import (
	"fmt"
	"log"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/scram"
)

type ScramAlgorithm string

const (
	SHA256 ScramAlgorithm = "SHA256"
	SHA512 ScramAlgorithm = "SHA512"
)

type Kafka struct {
	Brokers        []string       `json:"brokers" yaml:"brokers" toml:"brokers" env:"BROKERS" envSeparator:","`
	ScramAlgorithm ScramAlgorithm `json:"scram_algorithm" yaml:"scram_algorithm" toml:"scram_algorithm" env:"SCRAM_ALGORITHM"`
	Username       string         `json:"username" yaml:"username" toml:"username" env:"USERNAME"`
	Password       string         `json:"password" yaml:"password" toml:"password" env:"PASSWORD"`
	Debug          bool           `json:"debug" yaml:"debug" toml:"debug" env:"DEBUG"`

	mechanism sasl.Mechanism
}

func (k *Kafka) scramAlgorithm() scram.Algorithm {
	switch k.ScramAlgorithm {
	case SHA256:
		return scram.SHA256
	case SHA512:
		return scram.SHA512
	default:
		return scram.SHA256
	}
}

func (k *Kafka) Producer(topic string) *kafka.Writer {
	fmt.Printf("Kafka Producer: %s\n", topic)
	w := &kafka.Writer{
		Addr:                   kafka.TCP(k.Brokers...),
		Topic:                  topic,
		Balancer:               &kafka.LeastBytes{},
		AllowAutoTopicCreation: true,
		ErrorLogger:            kafka.LoggerFunc(log.Printf),
		Transport: &kafka.Transport{
			SASL: k.mechanism,
		},

		// RequiredAcks: kafka.RequireAll,
		// Async:        false,
		// Logger:       kafka.LoggerFunc(log.Printf),
		// Logger:       kafka.LoggerFunc(log.Printf),
	}
	if k.Debug {
		w.Logger = kafka.LoggerFunc(log.Printf)
	}
	return w
}

